package cn.dmp.tags

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis

object Tags4Buiness extends Tags {
  /**
    * 打标签的方法定义
    * 经纬度转geohash编码,然后根据geohash去redis中去查对应的商圈
    */
  override def makeTags(args: Any*): Map[String, Int] = {

    var map = Map[String, Int]()
    val row: Row = args(0).asInstanceOf[Row]
    val jedis: Jedis = args(1).asInstanceOf[Jedis]

    val lat: String = row.getAs[String]("lat")
    val long: String = row.getAs[String]("long")

    if (StringUtils.isNotEmpty(lat) && StringUtils.isNotEmpty(long)) {
      val lat2: Double = lat.toDouble
      val long2: Double = long.toDouble
      if (lat2 > 3 && lat2 < 54 && long2 > 73 && long2 < 136) {

        val geoHash: String = GeoHash.withCharacterPrecision(lat2, long2, 8).toBase32
        val business: String = jedis.get(geoHash)
        if (StringUtils.isNotEmpty(business)) business.split(",").foreach(bs => map += "BS" + bs -> 1)

      }
    }
    map
  }
}
